package com.rraj.common.mq;

import java.util.Date;
import java.util.Properties;

import javax.annotation.Resource;

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.SendCallback;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.order.OrderProducer;
import com.aliyun.openservices.ons.api.transaction.LocalTransactionExecuter;
import com.aliyun.openservices.ons.api.transaction.TransactionProducer;
import com.xiaoleilu.hutool.date.DateUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;

/**
 * Created by WangSheng on 2017/6/9.
 *
 * @author WangSheng
 * @date 2017/06/09
 */
@Component
public class MQClient {
    private static final Logger LOGGER = LoggerFactory.getLogger(MQClient.class);

    /**
     * 普通消息生产者
     */
    @Resource(name = "defaultProducer")
    @Lazy
    private Producer defaultProducer;
    /**
     * 全局顺序消息生产者
     */
    @Resource(name = "orderAllProducer")
    @Lazy
    private OrderProducer orderAllProducer;
    /**
     * 分区顺序消息生产者
     */
    @Resource(name = "orderShardingProducer")
    @Lazy
    private OrderProducer orderShardingProducer;
    /**
     * 事务消息生产者
     */
    @Resource(name = "transactionProducer")
    @Lazy
    private TransactionProducer transactionProducer;

    /**
     * 同步发送普通消息
     *
     * @param producerMsg 消息内容
     * @return 发送结果
     * @throws Exception 异常
     */
    public boolean sendCommonMessageSync(MQProducerMsg producerMsg) throws Exception {
        Message msg = buildMessage(producerMsg);
        SendResult sendResult = defaultProducer.send(msg);
        LOGGER.info(DateUtil.now() + " 同步发送普通消息." + sendResult + "," + new String(msg.getBody(), "UTF-8"));
        return true;
    }

    /**
     * 异步发送普通消息
     *
     * @param producerMsg  消息内容
     * @param sendCallback 异步调用类
     * @return 发送结果
     * @throws Exception 异常
     */
    public boolean sendCommonMessageAsync(MQProducerMsg producerMsg, SendCallback sendCallback) throws Exception {
        Message msg = buildMessage(producerMsg);
        defaultProducer.sendAsync(msg, sendCallback);
        LOGGER.info(DateUtil.now() + " 异步发送普通消息. topic=" + msg.getTopic() + ", msgId=" + msg.getMsgID() + "," + new String(msg.getBody(),
            "UTF-8"));
        return true;
    }

    /**
     * 单向发送普通消息
     *
     * @param producerMsg 消息内容
     * @return 发送结果
     * @throws Exception 异常
     */
    public boolean sendCommonMessageOneway(MQProducerMsg producerMsg) throws Exception {
        Message msg = buildMessage(producerMsg);
        defaultProducer.sendOneway(msg);
        LOGGER.info(DateUtil.now() + " 单向发送普通消息. topic=" + msg.getTopic() + ", msgId=" + msg.getMsgID() + "," + new String(msg.getBody(),
            "UTF-8"));
        return true;
    }

    /**
     * 发送定时消息
     *
     * @param producerMsg 消息内容
     * @param date        消息发送的时间点
     * @return 发送结果
     * @throws Exception 异常
     */
    public boolean sendTimingMessage(MQProducerMsg producerMsg, Date date) throws Exception {
        Message msg = buildMessage(producerMsg);
        msg.setStartDeliverTime(date.getTime());
        SendResult sendResult = defaultProducer.send(msg);
        LOGGER.info(DateUtil.now() + " 发送定时消息." + sendResult + "," + new String(msg.getBody(), "UTF-8"));
        return true;
    }

    /**
     * 发送延时消息
     *
     * @param producerMsg 消息内容
     * @param delayTime   延长的时间
     * @return 发送结果
     * @throws Exception 异常
     */
    public boolean sendDelayMessage(MQProducerMsg producerMsg, long delayTime) throws Exception {
        Message msg = buildMessage(producerMsg);
        msg.setStartDeliverTime(System.currentTimeMillis() + delayTime);
        SendResult sendResult = defaultProducer.send(msg);
        LOGGER.info(DateUtil.now() + " 发送延时消息." + sendResult + "," + new String(msg.getBody(), "UTF-8"));
        return true;
    }

    /**
     * 发送事务消息
     *
     * @param producerMsg              消息内容
     * @param localTransactionExecuter 本地事务执行器
     * @param transactionChecker       消息回查器
     * @param obj                      未知
     * @return 发送结果
     * @throws Exception 异常
     */
    public boolean sendTransactionMessage(MQProducerMsg producerMsg, LocalTransactionExecuter localTransactionExecuter,
                                          String transactionChecker,
                                          Object obj) throws Exception {
        Message msg = buildMessage(producerMsg);
        Properties properties = new Properties();
        properties.setProperty(MQConstant.TRANSACTION_TYPE_KEY, transactionChecker);
        msg.setUserProperties(properties);
        SendResult sendResult = transactionProducer.send(msg, localTransactionExecuter, obj);
        LOGGER.info(DateUtil.now() + " 发送事务消息." + sendResult + "," + new String(msg.getBody(), "UTF-8"));
        return true;
    }

    /**
     * 发送全局顺序消息
     *
     * @param producerMsg 消息内容
     * @return 发送结果
     * @throws Exception 异常
     */
    public boolean sendOrderAllMessage(MQProducerMsg producerMsg) throws Exception {
        Message msg = buildMessage(producerMsg);
        SendResult sendResult = orderAllProducer.send(msg, MQConstant.SHARDING_ALL_ORDER_TEST_TOPIC1_SHARDING1);
        LOGGER.info(DateUtil.now() + " 发送全局顺序消息." + sendResult + "," + new String(msg.getBody(), "UTF-8"));
        return true;
    }

    /**
     * 发送分区顺序消息
     *
     * @param producerMsg 消息内容
     * @return 发送结果
     * @throws Exception 异常
     */
    public boolean sendOrderShardingMessage(MQProducerMsg producerMsg, String shardingKey) throws Exception {
        Message msg = buildMessage(producerMsg);
        SendResult sendResult = orderShardingProducer.send(msg, shardingKey);
        LOGGER.info(DateUtil.now() + " 发送分区顺序消息." + sendResult + "," + new String(msg.getBody(), "UTF-8"));
        return true;
    }

    /**
     * 组装消息
     *
     * @param producerMsg
     * @return
     */
    private Message buildMessage(MQProducerMsg producerMsg) {
        Message msg = new Message(
            producerMsg.getTopic(),
            producerMsg.getTag(),
            producerMsg.getMsgBody());
        msg.setKey(producerMsg.getKey());
        return msg;
    }

}
